package com.app.kafka

import java.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import scala.collection.JavaConversions.mapAsScalaMap

object RedisDataProcess {

  def selectOffsetsFromRedisDatabase(topics:Array[String],group:String): Map[TopicPartition, Long] = {

    var offsetsMap:Map[TopicPartition, Long] = Map()

      for(topic<-topics){
        val m: util.Map[String, String] = RedisUtil.getRedisInstance.hgetAll(s"${topic}_${group}")

        val smap: scala.collection.mutable.Map[String, String] = m

        val map: Map[TopicPartition, Long] = smap.map(x => {
          val partition = x._1.toInt
          val offset = x._2.toLong
          new TopicPartition(topic, partition) -> offset
        }).toMap

        offsetsMap.++(map)
      }

    RedisUtil.closeRedis()
    offsetsMap
  }


  def updateOffsetsInRedisDatabase(offsetRanges: Array[OffsetRange], group: String): Unit = {
    for (offsetRange <- offsetRanges) {
      val topic = offsetRange.topic
      val partition = offsetRange.partition
      val offset = offsetRange.untilOffset

      RedisUtil.getRedisInstance.hset(s"${topic}_${group}", partition.toString, offset.toString)

      RedisUtil.closeRedis()
    }
  }
}